package qy.qyalgotrader.ts.barfeed;

import java.nio.file.Path;
import java.time.Duration;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.LocalTime;
import java.time.ZonedDateTime;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.OptionalDouble;
import java.util.OptionalInt;
import java.util.concurrent.*;
import java.util.function.Predicate;

import org.apache.commons.collections4.MapUtils;
import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.commons.math3.util.Precision;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.zeroc.Ice.ObjectPrx;

import it.unimi.dsi.fastutil.doubles.DoubleArrayList;
import qy.jalgotrade.bar.Bar;
import qy.jalgotrade.bar.Bars;
import qy.jalgotrade.bar.BasicBar;
import qy.jalgotrade.bar.Bar.Frequency;
import qy.jalgotrade.barfeed.BaseBarFeed;
import qy.jalgotrade.utils.CommonUtils;
import qy.jalgotrade.utils.resamplebase.TimeRange;
import qy.qyalgotrader.ts.TsDataUtils;
import qy.qyalgotrader.utils.CommonDataUtils;
import qy.qyalgotrader.utils.Constants;
import qy.qyalgotrader.utils.Helper;
import qy.qyalgotrader.utils.Services;
import qy.qyalgotrader.utils.apiservice.ts.TsData;
import qy.qyalgotrader.utils.apiservice.ts.TusharePrx;
import qy.qyalgotrader.wind.WindDataUtils;
import tech.tablesaw.api.DateTimeColumn;
import tech.tablesaw.api.DoubleColumn;
import tech.tablesaw.api.Row;
import tech.tablesaw.api.Table;
import tech.tablesaw.columns.Column;

/**
 * @author qy
 *
 */
public class TushareLiveFeed extends BaseBarFeed {

	private static final Logger __log = LoggerFactory.getLogger(TushareLiveFeed.class);

	/**
	 * timeout: 10, timeunit: {@link TimeUnit#MILLISECONDS MILLISECONDS)
	 */
	private static final Pair<Long, TimeUnit> QUEUE_POLL_TIMEOUT = Pair.of(10L, TimeUnit.MILLISECONDS);

	/**
	 * 分笔成交 采样间隔: 3s (中国 A 股 Level I)
	 */
	public static final Pair<Long, TimeUnit> TS_INQUERY_PERIOD_LEVEL_I = Pair
			.of((long) Frequency.LEVEL_I_SNAPSHOT_STOCK_A.getValue(), TimeUnit.SECONDS);

	/**
	 * 分笔成交 采样间隔: 6s (低负载)
	 */
	public static final Pair<Long, TimeUnit> TS_INQUERY_PERIOD_LOW_LOAD = Pair
			.of((long) Frequency.TICK_MONITOR_HIGH_LOAD.getValue(), TimeUnit.SECONDS);

	/**
	 * 
	 * @author c-geo
	 *
	 */
	private enum EventType {
		ON_BARS
	}

	/**
	 *
	 */
	private enum IntradayConst {
		PRE_CLOSE, TODAY_OPEN,
		PRE_POSITION
	}

	/**
	 * 
	 * @author c-geo
	 *
	 */
	public static class TickDataSeries {

		private final List<LocalDateTime> __dateTimes;

		private final DoubleArrayList __priceDS;

		private final DoubleArrayList __volumeDS;

		private final DoubleArrayList __amountDS;

		private final Map<String, DoubleArrayList> __extraDSs;

		private double __accuVolume;

		private double __accuAmount;

		private double __intradayVwap;

		private boolean __1stBar;

		private final Map<IntradayConst, Double> __intradayConsts;

		private double __intradayHigh;

		private double __intradayLow;

		private final Frequency __barFreq;

		private final boolean __isIndex;

		/**
		 * 
		 */
		public TickDataSeries(Frequency barFreq, boolean isIndex) {

			__priceDS = new DoubleArrayList();
			__volumeDS = new DoubleArrayList();
			__amountDS = new DoubleArrayList();
			__dateTimes = new ArrayList<>();
			__extraDSs = new HashMap<>();
			__accuVolume = Double.NaN;
			__accuAmount = Double.NaN;
			__intradayVwap = Double.NaN;
			__1stBar = true;
			__intradayConsts = MapUtils.putAll(new HashMap<>(), new Object[][] {
					{ IntradayConst.PRE_CLOSE, Double.NaN},
					{ IntradayConst.PRE_POSITION, Double.NaN},
					{ IntradayConst.TODAY_OPEN, Double.NaN},
			});
			__intradayHigh = Double.NaN;
			__intradayLow = Double.NaN;
			__barFreq = barFreq;
			__isIndex = isIndex;
		}

		/**
		 * 
		 */
		public void reset() {

			__priceDS.clear();
			__volumeDS.clear();
			__amountDS.clear();
			__dateTimes.clear();
			__extraDSs.forEach((extraField, dal) -> {
				dal.clear();
			});
		}

		/**
		 * 
		 * @return __dateTimes
		 */
		public List<LocalDateTime> getDateTimes() {

			return __dateTimes;
		}

		/**
		 * 
		 * @return __priceDS
		 */
		public DoubleArrayList getPriceDS() {

			return __priceDS;
		}

		/**
		 * 
		 * @return __volumeDS
		 */
		public DoubleArrayList getVolumeDS() {

			return __volumeDS;
		}

		/**
		 * 
		 * @return __amountDS
		 */
		public DoubleArrayList getAmountDS() {

			return __amountDS;
		}

		/**
		 * 
		 * @return __extraDSs
		 */
		public Map<String, DoubleArrayList> getExtraDSs() {

			return __extraDSs;
		}

		/**
		 * 
		 * @param extraField extraField
		 * @return __extraDSs
		 */
		public DoubleArrayList getExtraDS(String extraField) {

			return __extraDSs.get(extraField);
		}

		/**
		 *
		 * @return __barFreq
		 */
		public Frequency getBarFreq() {

			return __barFreq;
		}

		/**
		 *
		 * @return __isIndex
		 */
		public boolean isIndex() {

			return __isIndex;
		}

		/**
		 * 
		 * @return __priceDS.size() == 0
		 */
		public boolean empty() {

			return __priceDS.size() == 0;
		}

		/**
		 * 
		 * @return __priceDS.size()
		 */
		public int size() {

			return __priceDS.size();
		}

		/**
		 * pre_close, intraday_open, ... 日内行情常量:
		 */
		public void setIntradayConst(IntradayConst name, Double value) {

			__intradayConsts.put(name, value);
		}

		/**
		 * 
		 * @param dateTime dateTime
		 * @param price price
		 * @param volume volume
		 * @param amount amount
		 */
		public void appendTick(LocalDateTime dateTime, double price, double volume, double amount) {

			appendTick(dateTime, price, volume, amount, Collections.emptyMap());
		}

		/**
		 * 
		 * @param dateTime dateTime
		 * @param price price
		 * @param volume volume
		 * @param amount amount
		 * @param extras extras
		 */
		public void appendTick(LocalDateTime dateTime, double price, double volume, double amount,
				Map<String, Double> extras) {

			__dateTimes.add(dateTime);
			__priceDS.add(price);
			__volumeDS.add(volume);
			__amountDS.add(amount);

			for (String extraField : extras.keySet()) {
				if (!__extraDSs.containsKey(extraField)) {
					__extraDSs.put(extraField, new DoubleArrayList());
				}
				__extraDSs.get(extraField).add((double) extras.get(extraField));
			}
		}

		/**
		 * 
		 * @param dateTime dateTime
		 * @param frequency frequency
		 * @return Bar
		 * @throws Exception Exception
		 */
		public Bar buildBar(LocalDateTime dateTime, Frequency frequency) throws Exception {

			try {
				double open = __priceDS.getDouble(0);
				double high = CommonUtils.max(__priceDS);
				if (Double.isNaN(__intradayHigh) || high > __intradayHigh) {
					__intradayHigh = high;
				}
				double low = CommonUtils.min(__priceDS);
				if (Double.isNaN(__intradayLow) || low < __intradayLow) {
					__intradayLow = low;
				}
				double close = __priceDS.getDouble(__priceDS.size() - 1);
				double volume = CommonUtils.sum(__volumeDS);
				double amount = CommonUtils.sum(__amountDS);
				// TODO: 为保持与 min_1 频率基础历史行情数据计算逻辑一致, 指数 intradayVwap 计算逻辑应统一为: close 按 amount 加权平均
				if (__1stBar) {
					__accuVolume = volume;
					__accuAmount = amount;
					if (__isIndex) {
						__intradayVwap = close;
					} else {
						__intradayVwap = !CommonUtils.fuzzyEquals(__accuVolume, 0) ? __accuAmount / __accuVolume : Double.NaN;
					}
				} else {
					__accuVolume += volume;
					__accuAmount += amount;
					// 日内 vwap:
					if (__isIndex) {
						__intradayVwap = !CommonUtils.fuzzyEquals(__accuVolume, 0) ?
								(__intradayVwap * (__accuAmount - amount) + close * amount) / __accuAmount : Double.NaN;
					} else {
						__intradayVwap = !CommonUtils.fuzzyEquals(__accuVolume, 0) ? __accuAmount / __accuVolume : Double.NaN;
					}
				}
				// 涨幅:
				double pctChangeIntraDay = !Double.isNaN(__intradayConsts.get(IntradayConst.PRE_CLOSE)) ?
						(close - __intradayConsts.get(IntradayConst.PRE_CLOSE)) / __intradayConsts.get(IntradayConst.PRE_CLOSE) : Double.NaN;
				// 振幅:
				double rangeIntraDay = !Double.isNaN(__intradayHigh) && !Double.isNaN(__intradayLow) ?
						(__intradayHigh - __intradayLow) / __intradayLow : Double.NaN;

				// @formatter:off
				return new BasicBar(CommonDataUtils.toMarketDatetime(dateTime), open, high, low, close, volume, close,
						frequency, MapUtils.putAll(new HashMap<>(), new Object[][] {
								{ Constants.BAR_FIELD_AMOUNT, amount },
								{ Constants.BAR_FIELD_VWAP_INTRADAY, __intradayVwap },
								{ Constants.BAR_FIELD_PCT_CHANGE_INTRADAY, pctChangeIntraDay },
								{ Constants.BAR_FIELD_RANGE_INTRADAY, rangeIntraDay }
							}));
				// @formatter:on
			} finally {
				__1stBar = false;
			}
		}

		/*
		 * (non-Javadoc)
		 * @see java.lang.Object#toString()
		 */
		@Override
		public String toString() {

			List<Column<?>> cols = new ArrayList<>();
			cols.add(DateTimeColumn.create(Constants.DATE_TIME, __dateTimes));
			cols.add(DoubleColumn.create(Constants.PRICE, __priceDS.toDoubleArray()));
			cols.add(DoubleColumn.create(Constants.BAR_FIELD_VOLUME, __volumeDS.toDoubleArray()));
			cols.add(DoubleColumn.create(Constants.BAR_FIELD_AMOUNT, __amountDS.toDoubleArray()));
			__extraDSs.forEach((extraField, dal) -> {
				cols.add(DoubleColumn.create(extraField, dal.toDoubleArray()));
			});
			Table df = Table.create(cols);
			return df.toString();
		}
	}

	private class TushareLiveFeedPoller implements Runnable {

		private final Frequency __frequency;

		/**
		 * 当前 Bar 的 TimeRange:
		 */
		private TimeRange __currRange;

		/**
		 * 在当前 Bar 的 TimeRange 范围内, 第几次获取 ticks?
		 */
		private int __ticksPullCount;

		/**
		 * ZoneId: "UTC"
		 */
		private ZonedDateTime __nextBarClose;

		private final Map<String, TickDataSeries> __tickDSDict;

		private final Map<String, LocalDateTime> __lastQuoteTimeDict;

		/**
		 * 记录上一次的 Ticks: {identifier: Tick}
		 */
		private final Map<String, Row> __lastTickDict;

		private Pair<Long, TimeUnit> __pollInterval;

		private boolean __stopped;

		public TushareLiveFeedPoller(Frequency frequency, Pair<Long, TimeUnit> pollInterval) {

			__frequency = frequency;
			__tickDSDict = new HashMap<>();
			__lastQuoteTimeDict = new HashMap<>();
			__lastTickDict = new HashMap<>();
			__pollInterval = pollInterval;

			for (String identifier : __identifiers) {
				__tickDSDict.put(identifier, new TickDataSeries(__frequency, CommonDataUtils.isSseSzseIndex(identifier)));
			}

			__stopped = false;

			__currRange = null;
			__ticksPullCount = 0;
			__nextBarClose = null;

			__updateNextBarClose();
		}

		/**
		 * 
		 */
		public boolean stopped() {

			return __stopped;
		}

		/**
		 * 
		 */
		public void stop() {

			__stopped = true;
		}

		/*
		 * 间隔调用 tushare.get_realtime_quotes() 获取 A 股分笔成交行情, 在周期结束的
		 * NEXT_CALL 事件时, 生成 Bars, 插入 行情事件队列, 并通知相关的 订阅者 (策略):
		 * (non-Javadoc)
		 * @see java.lang.Runnable#run()
		 */
		@Override
		public void run() {

			__log.info("TushareLiveFeedPoller thread started.");

			while (!__stopped) {
				try {
					__wait();
					if (!__stopped) {
						doCall();
					}
				} catch (Exception e) {
					__log.error("Unhandled exception: {}", ExceptionUtils.getStackTrace(e));
				}
			}

			__log.info("TushareLiveFeedPoller thread finished.");
		}

		/**
		 * 
		 */
		private void __updateNextBarClose() {

			__currRange = TimeRange.buildRange(CommonDataUtils.utcNow(), __frequency);
			__nextBarClose = __currRange.getEnding();
		}

		/**
		 * 
		 */
		public void setPollInterval(Pair<Long, TimeUnit> pollInterval) {

			__pollInterval = pollInterval;
		}

		/**
		 * 
		 * @return __currRange
		 */
		public TimeRange getCurrentRange() {

			return __currRange;
		}

		/**
		 * 
		 * @return __nextBarClose
		 */
		public ZonedDateTime getNextCallDateTime() {

			return __nextBarClose;
		}

		/**
		 * doCall() 实现, 生成 (ON_BARS, Bars) 事件是在 周期 的 结束 时间点, 即每个 Bar 的时间为 Close 时间:
		 */
		public void doCall() {

			// 本次生成 (ON_BARS, Bars) 事件中 Bar 的 时间:
			ZonedDateTime barDateTime = __currRange.getBeginning();
			ZonedDateTime barEndTime = __currRange.getEnding();

			__updateNextBarClose();

			// 非交易时间不生成 Bars:
			if (!CommonDataUtils.isSseSzseTickDatetimeInBarTimeRange(__identifiers.get(0), CommonDataUtils.toMarketDatetime(barDateTime).toLocalDateTime(),
					OptionalInt.empty(), Optional.empty(), Optional.empty()) &&
					!CommonDataUtils.isSseSzseTickDatetimeInBarTimeRange(__identifiers.get(0), CommonDataUtils.toMarketDatetime(barEndTime).toLocalDateTime(),
							OptionalInt.empty(), Optional.empty(), Optional.empty())) {
				return;
			}

			Map<String, Bar> barDict = new HashMap<>();
			for (String identifier : __identifiers) {
				try {
					if (!__tickDSDict.get(identifier).empty()) {
						if (__log.isDebugEnabled()) {
							__log.debug("Building Bar for {} ({}), ticks: \n{}", identifier,
							        CommonDataUtils.toMarketDatetime(barDateTime), __tickDSDict.get(identifier));
						}
						Bar bar = __tickDSDict.get(identifier).buildBar(
								CommonDataUtils.toMarketDatetime(barDateTime).toLocalDateTime(), __frequency);
						if (__log.isDebugEnabled()) {
							__log.debug("Bar ({}) ok: \n{}", identifier, bar);
						}
						barDict.put(identifier, bar);
					} else {  // empty bar:
						if (__genEmptyBars) {
							Bar lastBar = null;
							Pair<EventType, Object> e = __deQueue.peekLast();
							if (e != null && e.getLeft() == EventType.ON_BARS) {
								Bars bars = (Bars) e.getRight();
								lastBar = bars.getBar(identifier);
							}
							if (lastBar != null && barDateTime.compareTo(lastBar.getDateTime()) > 0) {
								Bar currBar = CommonDataUtils.genEmptyBarBy(CommonDataUtils.toMarketDatetime(barDateTime), lastBar);
								barDict.put(identifier, currBar);
							}
						}
					}
				} catch (Exception e) {
					__log.error("buildBar({}, {}, {}) for {} error: {}", CommonDataUtils.toMarketDatetime(barDateTime),
					        __tickDSDict.get(identifier), __frequency, identifier, ExceptionUtils.getStackTrace(e));
				}
			}

			if (barDict.size() > 0) {
				try {
					Bars bars = new Bars(barDict);
					__deQueue.put(Pair.of(EventType.ON_BARS, bars));
				} catch (Exception e) {
					__log.error("Exception occured: {}", ExceptionUtils.getStackTrace(e));
				}
			}
		}

		/**
		 * 在 NEXT_CALL (产生 ON_BAR 事件) 之前, 每间隔 self._wsqPullInterval, 调用一次 __getTushareTickData():
		 */
		private void __wait() {

			__ticksPullCount = 0;
			// first reset ticks info in one cycle, maybe we need save it if NO quotation in this period
			for (String identifier : __identifiers) {
				__tickDSDict.get(identifier).reset();
			}
			// 跳过非交易时间:
			if (CommonDataUtils.isIllegalSseSzseTickDatetime(__identifiers.get(0),
					CommonDataUtils.toMarketDatetime(__currRange.getBeginning()).toLocalDateTime(),
					OptionalInt.empty(), Optional.empty(), Optional.empty()) &&
					CommonDataUtils.isIllegalSseSzseTickDatetime(__identifiers.get(0),
							CommonDataUtils.toMarketDatetime(__currRange.getEnding()).toLocalDateTime(),
							OptionalInt.empty(), Optional.empty(), Optional.empty())) {
				try {
					// 防止 CPU 空转高占用:
					Thread.sleep(25);
				} catch (InterruptedException e) {
					__log.error("InterruptedException: {}", ExceptionUtils.getStackTrace(e));
				}
				return;
			}

			ZonedDateTime nextCallTime = getNextCallDateTime();

			// TODO: 首次调用, 先 sleep() 到最近的一个标准 Tick 时间 slot:
			/*
			 * main loop: 阻塞直到 nextCallTime
			 */
			ZonedDateTime utcNow = CommonDataUtils.utcNow();
			while (!__stopped && utcNow.compareTo(nextCallTime) < 0) {

				if (__log.isDebugEnabled()) {
					__log.debug("utcnow(): {}, nextCallTime: {}", CommonDataUtils.toMarketDatetime(utcNow),
							CommonDataUtils.toMarketDatetime(nextCallTime));
				}

				ZonedDateTime startTime = CommonDataUtils.utcNow();
				// pull for tushare tick data:
				__getTushareTickData();
				__ticksPullCount++;
				ZonedDateTime endTime = CommonDataUtils.utcNow();

				Duration timeDiff = Duration.between(startTime, endTime);
				if (timeDiff.getSeconds() < __pollInterval.getLeft()) {
					long consumedMs = timeDiff.toMillis();
					// maxSleepMs: 若当前时间到 nextCallTime 不足一个 pollInterval:
					long maxSleepMs = Duration.between(endTime, nextCallTime).toMillis();
					long sleepMs = Math.min(__pollInterval.getRight().toMillis(__pollInterval.getLeft()) - consumedMs,
							maxSleepMs); // 防止睡过了头
					__log.debug("Time consumed getting Tick data: {} ms, sleep for {} ms", consumedMs, sleepMs);
					try {
						if (sleepMs > 0) {
							Thread.sleep(sleepMs);
						}
					} catch (InterruptedException e) {
						__log.error("InterruptedException occured: {}", ExceptionUtils.getStackTrace(e));
					}
				}
				utcNow = CommonDataUtils.utcNow();
			}
		}

		/**
		 * <pre>
		 * 调用一次 tushare.get_realtime_quotes(), 获取所有订阅标的最新的 实时行情, 并一一插入 self._tickDSDict 相应的 序列缓存 中;
		 * remark: tushare.get_realtime_quotes() 获取的 volume 和 amount 为当日累积的, 需计算得出 Tick 范围内的:
		 * </pre>
		 */
		private void __getTushareTickData() {

			try {
				TsData tData = __tushare.getRealtimeQuotes(__tsCodes.toArray(new String[0]));
				if (tData.errorCode != 0) {
					__log.error("Call tushare.get_realtime_quotes({}) with errors: {}", __identifiers, Helper.toString(tData));
				}

				// ticks for all codes:
				Table df = Helper.tsDataToDataFrameNoType(tData); // TODO: handle null or empty

				for (String identifier : __identifiers) {
					int codeIdx = __tsCodes.indexOf(TsDataUtils.windCodeToTsCode(identifier));
					if (codeIdx < 0 || ArrayUtils.indexOf(tData.codes, CommonDataUtils.windCodeToCode(identifier)) < 0) {
						throw new IllegalStateException(String.format("identifier %s(%s) NOT found in get_realtime_quotes response: %s",
								TsDataUtils.windCodeToTsCode(identifier), identifier, Helper.toString(tData)));
					}
					Row tick = df.row(codeIdx);

					String dtField = Constants.DATE;
					String tmField = Constants.TIME;
					String priceField = Constants.PRICE;
					String volField = Constants.VOLUME;
					String amtField = Constants.AMOUNT;

					if (__validTickData(identifier, tick)) {
						// @formatter:off
						if (__ticksPullCount == 0) {
							// 当前 freq TimeRange 内 首个 Ticks, 更新日内行情常量:
							// 前收盘:
							__tickDSDict.get(identifier).setIntradayConst(IntradayConst.PRE_CLOSE, Double.parseDouble(tick.getString(Constants.PRE_CLOSE)));
						}
						if (__lastTickDict.get(identifier) == null) {
							// TODO:
							__tickDSDict.get(identifier).appendTick(
									LocalDateTime.of(LocalDate.parse(tick.getString(dtField), Constants.DATE_FORMAT_STD),
											LocalTime.parse(tick.getString(tmField), Constants.TIME_FORMAT_STD)),
									Double.parseDouble(tick.getString(priceField)),
									Double.parseDouble(tick.getString(volField)),
									Double.parseDouble(tick.getString(amtField)));

							__log.debug("Initial Tick ({}) received: \n{}", identifier, tick);
						} else {
							Row lastTick = __lastTickDict.get(identifier);
							__tickDSDict.get(identifier).appendTick(
									LocalDateTime.of(LocalDate.parse(tick.getString(dtField), Constants.DATE_FORMAT_STD),
											LocalTime.parse(tick.getString(tmField), Constants.TIME_FORMAT_STD)),
									Double.parseDouble(tick.getString(priceField)),
									Double.parseDouble(tick.getString(volField))
											- Double.parseDouble(lastTick.getString(volField)),
					                Double.parseDouble(tick.getString(amtField))
					                		- Double.parseDouble(lastTick.getString(amtField)));
						}
						// @formatter:on
						// 更新记录的上一个有效 Tick:
						__lastTickDict.put(identifier, tick);
					}
				}

			} catch (Exception e) {
				__log.error("Tushare polling exception: {}", ExceptionUtils.getStackTrace(e));
			}
		}

		/**
		 * <p>
		 * 1. 确保 tickInfo 中的 time 一定晚于已记录的指定标的最新实时行情时间 (self._lastQuotationTime), 并更新之, 否则返回 Fale (放弃此条
		 * tickInfo);
		 * </p>
		 * <p>
		 * 2. 确保 tickInfo 中的 vol 一定大于已记录的上一个 tickInfo 的 vol, 否则返回 Fale (放弃此条 tickInfo);
		 * </p>
		 * <p>
		 * 3. 确保最新成交价格为合法.
		 * </p>
		 * 
		 * @param identifier identifier
		 * @param tick tick
		 * @return boolean
		 */
		private boolean __validTickData(String identifier, Row tick) {

			String dtField = Constants.DATE;
			String tmField = Constants.TIME;
			String volField = Constants.VOLUME;
			String preCloseField = Constants.PRE_CLOSE;
			String priceField = Constants.PRICE;
			LocalDateTime tickDateTime = LocalDateTime.of(LocalDate.parse(tick.getString(dtField), Constants.DATE_FORMAT_STD),
					LocalTime.parse(tick.getString(tmField), Constants.TIME_FORMAT_STD));

			if (__lastQuoteTimeDict.get(identifier) == null
			        || __lastQuoteTimeDict.get(identifier).compareTo(tickDateTime) < 0) {
				__lastQuoteTimeDict.put(identifier, tickDateTime);
			} else {
				return false;
			}

			if (__lastTickDict.get(identifier) != null
			        && __lastTickDict.get(identifier).getString(volField).equals(tick.getString(volField))) {
				return false;
			}
			// TODO: 精确到具体品种的涨跌停板:
			// @formatter:off
			return !(Double.parseDouble(tick.getString(priceField)) > Precision.round(Double.parseDouble(tick.getString(preCloseField))
					* Constants.PRICE_LMT_CN_STOCK_A_UP, Constants.PRICE_SIG_DIGITS_CN_STOCK_A)) &&
					!(Double.parseDouble(tick.getString(priceField)) < Precision.round(Double.parseDouble(tick.getString(preCloseField))
							* Constants.PRICE_LMT_CN_STOCK_A_DOWN, Constants.PRICE_SIG_DIGITS_CN_STOCK_A));
			// @formatter:on
		}
	}

	/**
	 * 必须是 WindCode:
	 */
	private final List<String> __identifiers;

	/**
	 * tushare code:
	 * e.g.: 601318, hs300, ...
	 */
	private final List<String> __tsCodes;

	private final Path __hisTickCsvBase;

	private final Path __hisMinCsvBase;

	private final LocalDate __today;

	private final LocalDate __currTd;

	private final boolean __genEmptyBars;

	/**
	 * Tushare:
	 */
	private final TusharePrx __tushare;

	/**
	 * self.__queue: 行情事件队列, 存放 (TuShareBarFeedThread.ON_BARS, Bars):
	 */
	private final BlockingDeque<Pair<EventType, Object>> __deQueue;

	private final TushareLiveFeedPoller __poller;

	private final Thread __thread;

	/**
	 * 
	 * @param identifiers identifiers
	 * @param frequency frequency
	 * @param maxLenP default: 0
	 * @param replayDaysP default: -1
	 * @param hisTickCsvBaseP default: CommonDataUtils.TS_TICK_CSV_BASE_DEFAULT
	 * @param hisMinCsvBaseP default: CommonDataUtils.TS_HIS_DATA_BASE_DIR_DEFAULT
	 * @param pollIntervalP default: TS_INQUERY_PERIOD_LEVEL_I
	 * @param genEmptyBarsP default: true
	 * @param useActiveModeP default: true
	 * @param slowReplayModeP default: false
	 * @throws Exception Exception
	 */
	// @formatter:off
	public TushareLiveFeed(List<String> identifiers, Frequency frequency,
			OptionalInt maxLenP, OptionalInt replayDaysP,
			Optional<Path> hisTickCsvBaseP,
			Optional<Path> hisMinCsvBaseP,
			Optional<Pair<Long, TimeUnit>> pollIntervalP, Optional<Boolean> genEmptyBarsP,
			Optional<Boolean> useActiveModeP) throws Exception {

		this(identifiers, frequency,
				maxLenP.orElse(0),
				replayDaysP.orElse(-1),
				hisTickCsvBaseP.orElse(CommonDataUtils.TS_TICK_CSV_BASE_DEFAULT),
				hisMinCsvBaseP.orElse(CommonDataUtils.TS_HIS_DATA_BASE_DIR_DEFAULT),
				pollIntervalP.orElse(TS_INQUERY_PERIOD_LEVEL_I),
				genEmptyBarsP.orElse(true),
				useActiveModeP.orElse(true));
		// @formatter:on
	}

	/**
	 * 
	 * @param identifiers identifiers
	 * @param frequency frequency
	 * @param maxLen maxLen
	 * @param replayDays replayDays
	 * @param hisTickCsvBase hisTickCsvBase
	 * @param hisMinCsvBase hisMinCsvBase
	 * @param pollInterval pollInterval
	 * @param useActiveMode useActiveMode
	 * @throws Exception Exception
	 */
	public TushareLiveFeed(List<String> identifiers, Frequency frequency, int maxLen, int replayDays,
			Path hisTickCsvBase, Path hisMinCsvBase, Pair<Long, TimeUnit> pollInterval, boolean genEmptyBars,
			boolean useActiveMode) throws Exception {

		super(frequency, maxLen);

		__identifiers = identifiers;
		__tsCodes = new ArrayList<>(__identifiers.size());
		__identifiers.forEach(identifier -> {
			__tsCodes.add(TsDataUtils.windCodeToTsCode(identifier));
		});
		// 历史 Tick, 历史 minute 行情 csv 文件根路径:
		__hisTickCsvBase = hisTickCsvBase;
		__hisMinCsvBase = hisMinCsvBase;

		__deQueue = new LinkedBlockingDeque<>();

		__today = LocalDate.now();
		__currTd = CommonDataUtils.nowTradingDate();
		__genEmptyBars = genEmptyBars;

		// Create a proxy to the Tushare
		ObjectPrx obj = Services.getInstance().createIceTushareProxy("Tushare");
		// Downcast the proxy to a Tushare proxy
		__tushare = TusharePrx.checkedCast(obj);

		__fillTodayAndHistoryBars(replayDays); // should run before polling thread start

		__poller = new TushareLiveFeedPoller(frequency, pollInterval);

		__thread = new Thread(__poller);

		for (String instrument : __identifiers) {
			registerInstrument(instrument);
		}
	}

	// barfeed.BaseBarFeed interface

	/* (non-Javadoc)
	 * @see qy.jalgotrade.barfeed.BaseBarFeed#getCurrentDateTime()
	 */
	@Override
	public ZonedDateTime getCurrentDateTime() {

//		return CommonDataUtils.utcNow();
		Bars currBars = getCurrentBars();
		if (currBars != null) {
			return currBars.getDateTime();
		} else {
			return CommonDataUtils.utcNow();
		}
	}

	/**
	 *
	 * @return __genEmptyBars
	 */
	public boolean getGenEmptyBars() {

		return __genEmptyBars;
	}

	/* (non-Javadoc)
	 * @see qy.jalgotrade.barfeed.BaseBarFeed#barsHaveAdjClose()
	 */
	@Override
	public boolean barsHaveAdjClose() {

		return false;
	}

	/* (non-Javadoc)
	 * @see qy.jalgotrade.barfeed.BaseBarFeed#getNextBars()
	 */
	@Override
	public Bars getNextBars() throws Exception {

		Bars ret = null;
		Pair<EventType, Object> item = __deQueue.poll(QUEUE_POLL_TIMEOUT.getLeft(), QUEUE_POLL_TIMEOUT.getRight());
		if (item != null) {
			if (item.getLeft() == EventType.ON_BARS) {
				ret = (Bars) item.getRight();
			} else {
				__log.error("Invalid event received: {} - {} !", item.getLeft(), item.getRight());
			}
			// 若 queue.get() 超时, 则返回 None:
		}
		return ret;
	}

	// observer.Subject interface

	/* (non-Javadoc)
	 * @see qy.jalgotrade.event.Subject#start()
	 */
	@Override
	public void start() {

		if (__thread.isAlive()) {
			throw new IllegalStateException("Already strated!");
		}
		// Start the thread that runs the client.
		__thread.start();
	}

	/* (non-Javadoc)
	 * @see qy.jalgotrade.event.Subject#stop()
	 */
	@Override
	public void stop() {

		// __thread.stop()
		__poller.stop();
	}

	/* (non-Javadoc)
	 * @see qy.jalgotrade.event.Subject#join()
	 */
	@Override
	public void join() {

		if (__thread.isAlive()) {
			try {
				__thread.join();
			} catch (InterruptedException e) {
				__log.warn("InterruptedException: {}", ExceptionUtils.getStackTrace(e));
			}
		}
	}

	/* (non-Javadoc)
	 * @see qy.jalgotrade.event.Subject#eof()
	 */
	@Override
	public boolean eof() {

		// return self.__thread.stopped()
		return __poller.stopped();
	}

	/* (non-Javadoc)
	 * @see qy.jalgotrade.event.Subject#peekDateTime()
	 */
	@Override
	public ZonedDateTime peekDateTime() {

		return null;
	}

	// TushareLiveFeed own interface
	/**
	 * 
	 * @param pollInterval pollInterval
	 */
	public void setPollInterval(Pair<Long, TimeUnit> pollInterval) {

		if (!__thread.isAlive()) {
			__poller.setPollInterval(pollInterval);
		} else {
			throw new IllegalStateException("can NOT set pollInterval when the poller thread is running.");
		}
	}

	/**
	 * 
	 * @return __today
	 */
	public LocalDate checkToday() {

		return __today;
	}

	/**
	 * 
	 * @return __currTd
	 */
	public LocalDate checkCurrTradingDate() {

		return __currTd;
	}

	/**
	 * 
	 * @return __poller.getCurrentRange()
	 */
	public TimeRange checkCurrTimeRange() {

		return __poller.getCurrentRange();
	}

	/**
	 * <pre>
	 * replayDays: 要获取的历史行情天数 (不包括当前交易日), 以支持历史行情重放:
	 *     -1 (或 <0): 不获取历史行情;
	 *     0: 获取当前交易日已经发生的行情;
	 *     >0: 要获取的历史行情的天数.
	 * </pre>
	 * 
	 * @param replayDays replayDays
	 * @throws Exception Exception
	 */
	private void __fillTodayAndHistoryBars(int replayDays) throws Exception {

		if (replayDays < 0) {
			replayDays = -1; // only allow -1 and >=0 integer value
		}
		if (replayDays == -1) {
			// 纯实时行情模式, 仅接收新行情
		} else if (replayDays == 0) {
			// replay today's quotation
			__fillTodayBars();
		} else {
			// 获取指定交易天数的历史行情, 和当前交易日已发生的行情, 使得 Feed 支持对其进行重放:
			__fillHistoryBars(replayDays);
			__fillTodayBars();
		}
	}

	/**
	 * 重放指定个数交易日的 历史行情:
	 * 
	 * @param replayDays replayDays
	 * @throws Exception Exception
	 */
	private void __fillHistoryBars(int replayDays) throws Exception {

		boolean inclZeroTd = __today.compareTo(__currTd) > 0;
		for (LocalDate day : CommonDataUtils.getTradingDays(__today, -replayDays, inclZeroTd)) {
			// barsDict: {证券代码: 历史行情 list (Bar)}
			LocalDateTime startDm = LocalDateTime.of(day, Constants.OPEN_TIME_AM);
			LocalDateTime endDm = LocalDateTime.of(day, Constants.CLOSE_TIME_PM);
			Map<String, List<Bar>> barsDict = new HashMap<>();
			for (String identifier : __identifiers) {
				List<Bar> barList = null;
				double priceRefVal;
				Table minsDf = null;
				if (CommonDataUtils.isSseSzseIndex(identifier)) {
					minsDf = WindDataUtils.readFromMinutesCsvWind(identifier, day, getFrequency(), Optional.of(WindDataUtils.WIND_HIS_DATA_BASE_DIR_DEFAULT),
							Optional.empty(), Optional.empty());
				} else {
					minsDf = CommonDataUtils.readFromMinutesCsvTs(identifier, day, getFrequency(), Optional.of(__hisMinCsvBase),
							Optional.empty(), Optional.empty());
				}
				// 直接读取 minutes csv 失败, 尝试读取 tick csv:
				if (minsDf == null || minsDf.isEmpty()) {
					Table ticksDf = null;
					if (CommonDataUtils.isSseSzseIndex(identifier)) {
						ticksDf = WindDataUtils.readFromTicksCsvWind(identifier, day, Optional.of(WindDataUtils.WIND_TICK_CSV_BASE_DEFAULT),
								Optional.empty(), Optional.empty());
					} else {
						ticksDf = CommonDataUtils.readFromTicksCsvTs(identifier, day, Optional.of(__hisTickCsvBase),
								Optional.empty(), Optional.empty());
					}
					// 直接读取 ticks csv 失败, 调用 tushare 接口获取:
					if (ticksDf == null || ticksDf.isEmpty()) {
						TsData tsData = __tushare.getTickData(TsDataUtils.windCodeToTsCode(identifier),
								day.format(Constants.DATE_FORMAT_STD), OptionalInt.empty(),
								OptionalDouble.of(0.5), Optional.of(Constants.TS_DEFAULT_GET_TICK_DATA_SRC));
						if (tsData != null && tsData.errorCode == 0 ) {
							ticksDf = Helper.tsDataToDataFrame(tsData);
							if (ticksDf != null && !ticksDf.isEmpty()) {
								minsDf = CommonDataUtils.sampleDataFrameTicksToMinuteTs(identifier, day, ticksDf, startDm, endDm, true);
								if (!(getFrequency() == Frequency.MINUTE || getFrequency() == Frequency.MINUTE_1)) {
									// 前收盘 = 当日首个 Tick Price - change
									priceRefVal = ticksDf.row(0).getDouble(Constants.PRICE) - ticksDf.row(0).getDouble(Constants.CHANGE);
									minsDf = CommonDataUtils.resampleDataFrameMinuteToTs(minsDf, startDm, endDm, getFrequency(), priceRefVal);
								}
								barList = CommonDataUtils.minutesDataFrameToBarListTs(minsDf, getFrequency(), identifier, day);
								if (barList != null) {
									barsDict.put(identifier, barList);
								}
							} else {
								barsDict.put(identifier, Collections.emptyList());
							}
						} else {
							barsDict.put(identifier, Collections.emptyList());
						}
					} else {
						// 处理 ticks DataFrame:
						if (ticksDf != null && !ticksDf.isEmpty()) {
							minsDf = CommonDataUtils.sampleDataFrameTicksToMinuteTs(identifier, day, ticksDf, startDm, endDm, true);
							if (!(getFrequency() == Frequency.MINUTE || getFrequency() == Frequency.MINUTE_1)) {
								// 前收盘 = 当日首个 Tick Price - change
								priceRefVal = ticksDf.row(0).getDouble(Constants.PRICE) - ticksDf.row(0).getDouble(Constants.CHANGE);
								minsDf = CommonDataUtils.resampleDataFrameMinuteToTs(minsDf, startDm, endDm, getFrequency(), priceRefVal);
							}
							barList = CommonDataUtils.minutesDataFrameToBarListTs(minsDf, getFrequency(), identifier, day);
							if (barList != null) {
								barsDict.put(identifier, barList);
							} else {
								barsDict.put(identifier, Collections.emptyList());
							}
						} else {
							barsDict.put(identifier, Collections.emptyList());
						}
					}
				} else {
					// 处理 minutes DataFrame:
					if (!(getFrequency() == Frequency.MINUTE || getFrequency() == Frequency.MINUTE_1)) {
						// 前收盘 = 当日首个 Bar close - change
						priceRefVal = minsDf.row(0).getDouble(Constants.BAR_FIELD_CLOSE) - minsDf.row(0).getDouble(Constants.CHANGE);
						minsDf = CommonDataUtils.resampleDataFrameMinuteToTs(minsDf, startDm, endDm, getFrequency(), priceRefVal);
					}
					barList = CommonDataUtils.minutesDataFrameToBarListTs(minsDf, getFrequency(), identifier, day);
					if (barList != null) {
						barsDict.put(identifier, barList);
					} else {
						barsDict.put(identifier, Collections.emptyList());
					}
				}
			}
			// 批量添加 bar list:
			__fillBars(day, barsDict);
		}
	}

	/**
	 * 重放当前交易日的 历史行情:
	 * @throws Exception Exception
	 */
	private void __fillTodayBars() throws Exception {

		LocalDate today = LocalDate.now();
		LocalTime now = LocalTime.now();
		if (CommonDataUtils.isTradingDay(today)) { // do nothing if holiday
			// todayBars: {证券代码: 当日历史行情 list (Bar)}
			Map<String, List<Bar>> todayBars = new HashMap<>();
			List<Bar> aBarList = null;
			for (String identifier : __identifiers) {
				try {
					LocalDateTime startDm = LocalDateTime.of(today, Constants.OPEN_TIME_AM);
					LocalDateTime endDm;
					TsData tickTsData;
					if (now.compareTo(Constants.REPLAY_TIME_START) > 0) {
						// 1. 已收盘 (至午夜前), (当日) 历史 Tick 行情已可获取: OPEN_TIME_AM ~ CLOSE_TIME_PM:
						endDm = LocalDateTime.of(today, Constants.CLOSE_TIME_PM);

						tickTsData = __tushare.getTickData(TsDataUtils.windCodeToTsCode(identifier),
								today.format(Constants.DATE_FORMAT_STD),
								OptionalInt.empty(), OptionalDouble.empty(),
								Optional.of(Constants.TS_DEFAULT_GET_TICK_DATA_SRC));

						Table ticksDf = Helper.tsDataToDataFrame(tickTsData);
						Table minsDf = CommonDataUtils.sampleDataFrameTicksToMinuteTs(identifier, today, ticksDf, startDm, endDm, true);
						if (!(getFrequency() == Frequency.MINUTE || getFrequency() == Frequency.MINUTE_1)) {
							// 前收盘 = 当日首个 Tick Price - change
							double priceRefVal = ticksDf.row(0).getDouble(Constants.PRICE) - ticksDf.row(0).getDouble(Constants.CHANGE);
							minsDf = CommonDataUtils.resampleDataFrameMinuteToTs(minsDf, startDm, endDm, getFrequency(), priceRefVal);
						}
						aBarList = CommonDataUtils.minutesDataFrameToBarListTs(minsDf, getFrequency(), identifier, today);
					} else if (now.compareTo(Constants.OPEN_TIME_AM) > 0) {
						// 2. 盘中: OPEN_TIME_AM ~ NOW:
						endDm = LocalDateTime.of(today, LocalTime.now());

						tickTsData = __tushare.getTodayTicks(TsDataUtils.windCodeToTsCode(identifier),
								OptionalInt.empty(), OptionalDouble.empty());

					} else {
						// 3. 其他时间段 (午夜至上午开盘前, 建议使用 CsvBarFeed): 不获取:
						todayBars.put(identifier, Collections.emptyList());
						continue;
					}

					Table ticksDf = Helper.tsDataToDataFrame(tickTsData);
					Table minsDf = CommonDataUtils.sampleDataFrameTicksToMinuteTs(identifier, today, ticksDf, startDm, endDm, true);
					if (!(getFrequency() == Frequency.MINUTE || getFrequency() == Frequency.MINUTE_1)) {
						// 前收盘 = 当日首个 Tick Price - change
						double priceRefVal = ticksDf.row(0).getDouble(Constants.PRICE) - ticksDf.row(0).getDouble(Constants.CHANGE);
						minsDf = CommonDataUtils.resampleDataFrameMinuteToTs(minsDf, startDm, endDm, getFrequency(), priceRefVal);
					}
					aBarList = CommonDataUtils.minutesDataFrameToBarListTs(minsDf, getFrequency(), identifier, today);

					if (aBarList != null) {
						todayBars.put(identifier, aBarList);
					}
				} catch (Exception e) {
					__log.error("Exception occured getting today ticks: {}", ExceptionUtils.getStackTrace(e));
				}
			}
			__fillBars(today, todayBars);
		}
	}

	/**
	 * 向 行情事件队列 中插入 (ON_BARS, bars) 事件, 实现历史行情重放 (对 TushareLiveFeed 进行迭代获取 行情 时, 将会依次取出):
	 * <br/>
	 * theBarsDict: {证券代码: 特定交易日的历史行情 barList}
	 * @param theBarsDict 其中 List&lt;Bar&gt; 均未非空
	 * @throws Exception Exception
	 */
	private void __fillBars(LocalDate theDate, Map<String, List<Bar>> theBarsDict) throws Exception {

		Map<String, Integer> allIdxesEnd = new HashMap<>(); // 所有 barList 的长度
		Map<String, Integer> currIdxes = new HashMap<>(); // 当前 index
		Map<String, ZonedDateTime> currDmColl = new HashMap<>(); // 当前 value (ZonedDateTime)
		// 简化调用方逻辑, 过滤 theBarsDict 中的 emptyList:
		Map<String, List<Bar>> filteredDict = new HashMap<>();
		theBarsDict.forEach((identifier, barList) -> {
			if (!barList.isEmpty()) {
				filteredDict.put(identifier, barList);
			}
		});
		if (filteredDict.isEmpty()) {
			return;
		}

		for (String e : filteredDict.keySet()) {
			allIdxesEnd.put(e, filteredDict.get(e).size());
			currIdxes.put(e, 0);
			currDmColl.put(e, filteredDict.get(e).get(currIdxes.get(e)).getDateTime());
		}
		ZonedDateTime currDm = Collections.min(currDmColl.values());

		Map<String, Bar> currBarDict = new HashMap<>();
		// 停止迭代条件: 所有 barList 的 index 均 reached end:
		Predicate<Object> cond = (obj) -> {
			for (Map.Entry<String, Integer> e : currIdxes.entrySet()) {
				if (e.getValue() < allIdxesEnd.get(e.getKey())) {
					return true;
				}
			}
			return false;
		};
		while (cond.test(null)) {
			for (Map.Entry<String, ZonedDateTime> e : currDmColl.entrySet()) {
				if (e.getValue().equals(currDm)) {
					String identifier = e.getKey();
					if (currIdxes.get(identifier) < allIdxesEnd.get(identifier)) {
						currBarDict.put(identifier, filteredDict.get(identifier).get(currIdxes.get(identifier)));
						// currIdxes 中相应的 index 后移:
						currIdxes.put(identifier, currIdxes.get(identifier) + 1);
					}
				}
			}
			// 更新 currDmColl 以及 currDm:
			currDmColl.clear();
			for (Map.Entry<String, Integer> e : currIdxes.entrySet()) {
				String identifier = e.getKey();
				if (currIdxes.get(identifier) < allIdxesEnd.get(identifier)) {
					currDmColl.put(identifier, filteredDict.get(identifier).get(e.getValue()).getDateTime());
				}
			}
			if (!currDmColl.isEmpty()) {
				currDm = Collections.min(currDmColl.values());
			}

			Bars bars = new Bars(currBarDict);
			__deQueue.put(Pair.of(EventType.ON_BARS, bars));
			currBarDict = new HashMap<>();
		}
	}
}
